package com.sh.rpc.registry;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSON;
import cn.hutool.json.JSONUtil;
import com.sh.rpc.confige.RegistryConfig;
import com.sh.rpc.model.ServiceMetaInfo;
import io.etcd.jetcd.*;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.watch.WatchEvent;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

public class EtcdRegistry implements Registry {


    private Client client;  // etcd客户端
    private KV kvClient;  // etcd KV客户端
    public static final Set<String> localRegisterNodeSet = new HashSet<>();  // 本地注册节点集合
    private RegistryServiceCache registryServiceCache = new RegistryServiceCache();  // 注册中心服务缓存
    private final Set<String> watchingKeySet = new HashSet<>();  // 正在监听的key集合


    /**
     * 根节点
     * 这是向 etcd 注册的根节点
     */
    private static final String ETCD_ROOT_PATH = "/rpc/";


    @Override
    public void init(RegistryConfig registryConfig) {
        // 初始化etcd客户端
        client = Client.builder()
                .endpoints(registryConfig.getAddress())    // "http://localhost:2380"
                .connectTimeout(Duration.ofMillis(registryConfig.getTimeout()))
                .build();
        kvClient = client.getKVClient();  // 初始化KV客户端
        // 启动心跳任务
        heartbeat();
    }

    /**
     * 向服务中心 etcd 注册服务信息
     * @param serviceMeteInfo
     * @throws Exception
     */
    @Override
    public void register(ServiceMetaInfo serviceMeteInfo) throws Exception {
        // 创建lease 和 KV 客户端
        /**
         * Lease: 租约 它代表了对某个资源的临时授权或使用期限。
         */
        Lease leaseClient = client.getLeaseClient();

        /**
         * 创建一个 30 秒的租约
         * 租约在 RPC 中通常是指一个节点（通常是客户端）
         * 从另一个节点（通常是服务器）获取的对特定资源的有限时间的使用权限。
         * 这里会是阻塞式操作   请确定你的服务器是开着的
         */
        long leaseId = leaseClient.grant(1000).get().getID();

        // 设置键值对                           名字 + 版本 + 域名 + 端口
        String registerKey = ETCD_ROOT_PATH + serviceMeteInfo.getServiceNodeKey();
        // StandardCharsets提供了对标准字符集的访问
        // ByteSequence是一个接口，通常用于表示字节序列
        // 将指定的字符串 registerKey 以 UTF-8 编码转换为 ByteSequence 类型的键
        ByteSequence key = ByteSequence.from(registerKey, StandardCharsets.UTF_8);
        // 将 serviceMeteInfo 对象转换为 JSON 字符串后，再以 UTF-8 编码转换为 ByteSequence 类型的值
        ByteSequence value = ByteSequence.from(JSONUtil.toJsonStr(serviceMeteInfo), StandardCharsets.UTF_8);
        // 创建租约并设置键值对   就是将 键值对和租约绑定
        // 创建一个 PutOption 的构建器，用于配置向 etcd 存储键值对的选项
        PutOption putOption = PutOption.builder()
                // 设置要关联的租约 ID，通过这个租约来管理键值对的有效期
                .withLeaseId(leaseId)
                // 构建出最终的 PutOption 对象
                .build();
        kvClient.put(key, value, putOption).get();
        // 添加节点信息
        localRegisterNodeSet.add(registerKey);
    }


    /**
     * 取消注册 服务端
     */
    @Override
    public void unRegister(ServiceMetaInfo serviceMeteInfo) throws Exception {
        String registerKey = ETCD_ROOT_PATH + serviceMeteInfo.getServiceNodeKey();
        kvClient.delete(ByteSequence.from(registerKey, StandardCharsets.UTF_8));
        localRegisterNodeSet.remove(registerKey);   // 删除本地节点信息
        // todo 删除缓存
    }

    /**
     * 服务发现
     *
     * @param serviceKey
     * @return
     */
    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
        // 发现前先看看缓存有没有
        List<ServiceMetaInfo> cachedServiceMetaInfoList = registryServiceCache.readCache();
        if (cachedServiceMetaInfoList != null) {
            return cachedServiceMetaInfoList;
        }
        // 前缀搜索  结尾一定要加 ‘/’
        String searchPrefix = ETCD_ROOT_PATH + serviceKey + '/';
        try {
            GetOption option = GetOption.builder().isPrefix(true).build();
            // 获取所有符合条件的键值对
            List<KeyValue> keyValues = kvClient.get(ByteSequence.from(searchPrefix, StandardCharsets.UTF_8), option)
                    .get()
                    .getKvs();
            // 解析服务消息
            List<ServiceMetaInfo> serviceMetaInfoList = keyValues.stream().map(
                            keyValue -> {
                                String key = keyValue.getKey().toString(StandardCharsets.UTF_8);
                                watch(key);
                                String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                                return JSONUtil.toBean(value, ServiceMetaInfo.class);
                            })
                    .collect(Collectors.toList());
            // 写入缓存
            registryServiceCache.writeCache(serviceMetaInfoList);
            return serviceMetaInfoList;
        } catch (Exception e) {
            throw new RuntimeException("获取服务列表失败" + e);
        }
    }


    /**
     * 节点下线
     */
    @Override
    public void destroy() {
        System.out.println("当前节点 etcd 下线");
        for (String key : localRegisterNodeSet) {
            // 删除节点
            try {
                kvClient.delete(ByteSequence.from(key, StandardCharsets.UTF_8)).get();
            } catch (Exception e) {
                e.printStackTrace();
                throw new RuntimeException(key + "下线失败");
            }
        }
        if (kvClient != null) {
            kvClient.close();
        }
        if (client != null) {
            client.close();
        }
    }

    /**
     * 心跳机制
     */
    @Override
    public void heartbeat() {
        // 每隔 10 秒执行一次
        CronUtil.schedule("0 */10 * * * *", new Task() {
            @Override
            public void execute() {
                System.out.println("开始续约");
                for (String key : localRegisterNodeSet) {
                    try {
                        List<KeyValue> keyValues = kvClient.get(ByteSequence.from(key, StandardCharsets.UTF_8))
                                .get()
                                .getKvs();
                        // 过期， 没有了，要重启才继续注册
                        if (CollUtil.isEmpty(keyValues)) {
                            continue;
                        }
                        // 续约  未过期
                        KeyValue keyValue = keyValues.get(0);
                        String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                        ServiceMetaInfo serviceMetaInfo = JSONUtil.toBean(value, ServiceMetaInfo.class);
                        register(serviceMetaInfo);
                    } catch (Exception e) {
                        throw new RuntimeException(key + "续约失败" + e);
                    }
                }

            }
        });
        CronUtil.setMatchSecond(true);  // 精确到秒
        CronUtil.start();       // 启动定时任务
    }


    /**
     * 监听  调用的时候监听
     *
     * @param serviceNodeKey
     */
    @Override
    public void watch(String serviceNodeKey) {
        Watch watchClient = client.getWatchClient();
        // 获取用于监听键值变化的客户端对象
        boolean newWatch = watchingKeySet.add(serviceNodeKey);
        // 将 serviceNodeKey 添加到 watchingKeySet 集合中。
        // 如果集合中原本不存在这个键，add 方法会返回 true，表示这是第一次对这个键进行监听；
        // 如果集合中已经存在这个键，会返回 false。
        if (newWatch) {
            // 如果是第一次对这个键进行监听，执行以下代码
            watchClient.watch(ByteSequence.from(serviceNodeKey, StandardCharsets.UTF_8),
                    response -> {
                        // 启动对特定键（serviceNodeKey）的监听，当这个键对应的值发生变化时，会触发这个回调函数。
                        // 参数 response 包含了监听到的事件信息
                        for (WatchEvent event : response.getEvents()) {
                            // 遍历监听到的事件
                            switch (event.getEventType()) {
                                case DELETE:
                                    // 如果监听到的事件类型是 DELETE，表示这个键对应的值被删除了
                                    // registryServiceCache.clearCache(serviceNodeKey);
                                    // 注释中的代码可能是清除特定键的缓存
                                    registryServiceCache = null;
                                    // 将 registryServiceCache 设为 null，表示进行了某种清理操作
                                    break;
                                case PUT:
                                default:
                                    // 如果监听到的事件类型是 PUT 或者其他默认情况，目前没有具体处理
                                    break;
                            }
                        }
                    });
        }
    }


}
